package com.disruptor.disruptorStudy.demo02;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import com.disruptor.disruptorStudy.demo01.Trade;
import com.lmax.disruptor.BusySpinWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.EventHandlerGroup;
import com.lmax.disruptor.dsl.ProducerType;

//菱形业务: 生产者生产数据,先交给c1,c2并行消费,消费完之后再交给c3进行消费
public class Test {
    @SuppressWarnings("unchecked")
    public static void main(String[] args) throws Exception {
        long beginTime = System.currentTimeMillis();
        int BUFFER_SIZE = 1024;
        //创建线程池
        ExecutorService executor = Executors.newFixedThreadPool(4);
        Disruptor<Trade> disruptor = new Disruptor<Trade>(
                () -> {
                    return new Trade();
                }, //事件工厂
                BUFFER_SIZE, executor,  //buffer大小   线程池
                // 生产模式(单生产者)   等待策略
                ProducerType.SINGLE, new BusySpinWaitStrategy());
        //菱形业务处理
        @SuppressWarnings("unchecked")
        EventHandlerGroup<Trade> handleEventsWith = disruptor.handleEventsWith(
                // 两个消费者并行消费数据    EventHandler接口实现类
                (event, sequence, endOfBatch) -> {
                    event.setName("侯征");
                },//设置名字
                (event, sequence, endOfBatch) -> {
                    event.setPrice(300);
                });
        //前两个消费之后,第三个再进行消费
        handleEventsWith.then((event, sequence, endOfBatch) -> {
            System.out.println("获取处理后的数据id:" + event.getId() +
                    " 姓名: " + event.getName() + " 价格:" + event.getPrice());
        });
        disruptor.start();//启动
        //生产者生产数据
        CountDownLatch latch = new CountDownLatch(1);//监听一个线程(生产者),等这个线程通知后再继续执行
        //生产者开启线程生产数据
        executor.submit(new TradePublisher(latch, disruptor));//单生产者

        latch.await();//等待生产者完事通知,防止没生产完数据,提前关闭
        disruptor.shutdown();
        executor.shutdown();
        System.out.println("总耗时:" + (System.currentTimeMillis() - beginTime));
    }

}
